package javaVersion.learn.stateProcess.keyStateDemo;

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;

/**
 * @program: myFlink
 * @description: 操作State 句柄(官网Java版案例)
 * @author: WincoMa
 * @create: 2020-07-24 15:49
 **/
public class KeyState_StateManager extends RichFlatMapFunction<Tuple2<Long,Long>,Tuple2<Long, Long>> {
    private transient ValueState<Tuple2<Long,Long>> sum;
    @Override
    public void flatMap(Tuple2<Long, Long> input, Collector<Tuple2<Long, Long>> out) throws Exception {
        //访问状态值
        Tuple2<Long, Long> tmpCurrentSum = sum.value();
        //如果以前没有使用过，则为空
        Tuple2<Long, Long> currentSum = new Tuple2<>();
        if (tmpCurrentSum != null) {
            currentSum = tmpCurrentSum;
        }else {
            currentSum = Tuple2.of(0L,0L);
        }

        // TODO: 2020/7/24 更新状态值逻辑
        //更新状态值,(使用新Tuple来接收新状态值)
        //更新逻辑：输入元素为：以相同key为分组的多个不同value。
        //          以输入的key为基准，在内部Tuple<key2,value2>，
        //                             对于每次更新，key2为计数，
        //                                           value2为：真实数据统计
        //                                                     都把输入数据的value值加入到
        //                                                     state的value2中
        Tuple2<Long, Long> newSum = new Tuple2<>();
        newSum.setFields(currentSum.f0 + 1,currentSum.f1+input.f1);
        //更新状态值
        sum.update(newSum);

        // TODO: 2020/7/24  如果计数达到2，则发出平均值并清除状态。
        if (newSum.f0 >=2) {
            Tuple2<Long, Long> outTuple = new Tuple2<>(input.f0,newSum.f1/newSum.f0);
            out.collect(outTuple);
            sum.clear();
        }
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        sum = getRuntimeContext().getState(
                new ValueStateDescriptor<Tuple2<Long, Long>>("average",TypeInformation.of(
                        new TypeHint<Tuple2<Long, Long>>() {
                })));
    }
}
